package queue

import (
	"cashcenter/internal/model"
	"cashcenter/utils/logx"
	"encoding/json"
	"github.com/pkg/errors"
	"github.com/streadway/amqp"
	"strconv"
	"sync"
)

var (
	ce <-chan amqp.Delivery
	cd <-chan amqp.Delivery
)
var wg sync.WaitGroup

func consume(xcg string) {
	var err error

	ce, err = MQChan.Consume(
		pre+RkExec,
		"ctpCashExec",
		true,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		logx.LogrusObj.Fatalln(errors.Wrap(err, "[QUEUE] can't get consume"))
	}
	cd, err = MQChan.Consume(
		pre+RkDead,
		"ctpCashDead",
		true,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		logx.LogrusObj.Fatalln(errors.Wrap(err, "[QUEUE] can't get consume"))
	}

	var consumers []func(string)
	consumers = append(consumers, exec)
	consumers = append(consumers, dead)

	wg.Add(len(consumers))

	for _, f := range consumers {
		go f(xcg)
	}
}

// =================================== consume logic ===================================

func exec(xcg string) {
	for msg := range ce {
		var err error
		var rl int

		// 获取重试次数
		trl, okt := msg.Headers["retry-left"]
		if !okt {
			rl = 0
		} else {
			var err error
			srl, oks := trl.(string)
			if !oks {
				rl = 0
			} else {
				rl, err = strconv.Atoi(srl)
				if err != nil {
					rl = 0
				}
			}
		}

		// 执行消息
		var data InCloseContent
		err = json.Unmarshal(msg.Body, &data)
		if err != nil {
			fail(xcg, msg.Body, rl, err)
			continue
		}
		err = model.UpdateCashIn(&model.CashIn{
			InId:  data.Uuid,
			State: data.State,
		})
		if err != nil {
			fail(xcg, msg.Body, rl, err)
			continue
		}
		// TODO 关闭支付订单 Close()

	}
}

func dead(xcg string) {
	for msg := range cd {

		// TODO log dead logic
		logx.LogrusObj.Printf("get a dead msg: %s\n", msg.Body)
	}
}

// =================================== republish logic ===================================

func retryMsg(exchange string, msgBody []byte, retryCount string) {
	err := MQChan.Publish(
		exchange,
		RkExec,
		false,
		false,
		amqp.Publishing{
			Headers:         amqp.Table{"retry-left": retryCount},
			ContentType:     "text/plain",
			ContentEncoding: "",
			Body:            msgBody,
		},
	)
	if err != nil {
		panic(err)
	}
}

func killMsg(exchange string, msgBody []byte) {
	err := MQChan.Publish(
		exchange,
		RkDead,
		false,
		false,
		amqp.Publishing{
			Headers:         nil,
			ContentType:     "text/plain",
			ContentEncoding: "",
			Body:            msgBody,
		},
	)
	if err != nil {
		panic(err)
	}
}

// =================================== publish logic ===================================

func PubInClose(uuid string) error {
	msg, _ := json.Marshal(InCloseContent{
		Uuid:  uuid,
		State: "closed",
	})
	err := MQChan.Publish(Xcg, "Wait", false, false, amqp.Publishing{
		Headers:         amqp.Table{"retry-left": "2"},
		ContentType:     "text/plain",
		ContentEncoding: "",
		Body:            msg,
	})

	return err
}

// =================================================================

func fail(xcg string, msg []byte, left int, reason error) {
	logx.LogrusObj.Errorf("[Queue] Get a exec msg: %s\n, but can't exec: %s", msg, reason.Error())
	if left >= 1 {
		retryMsg(xcg, msg, strconv.Itoa(left-1))
	} else {
		dMsg, _ := json.Marshal(DeadContent{
			Content: msg,
			Reason:  reason.Error(),
		})
		killMsg(xcg, dMsg)
	}
}

type InCloseContent struct {
	Uuid  string
	State string
}
type DeadContent struct {
	Content []byte
	Reason  string
}
